package com.mxw.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

public class RingBufferWorkerPoolFactory {

    //单例
    private static class SingletonHolder {
        static final RingBufferWorkerPoolFactory instance = new
                RingBufferWorkerPoolFactory();
    }

    private RingBufferWorkerPoolFactory() {
    }

    public static RingBufferWorkerPoolFactory getInstance() {
        return SingletonHolder.instance;
    }

    //需要生产者池和消费者池管理生产和消费者。
    private static Map<String, MessageProducer> producers = new
            ConcurrentHashMap<String, MessageProducer>();

    private static Map<String, MessageConsumer> consumers = new
            ConcurrentHashMap<String, MessageConsumer>();

    private RingBuffer<TranslatorDataWapper> ringBuffer;

    private SequenceBarrier sequenceBarrier;


    private WorkerPool<TranslatorDataWapper> workerPool;


    //初始化ProducerType生产者类型，是多生产还是单生产。MessageConsumer[]多消费者
    public void initAndStart(ProducerType type, int bufferSize, WaitStrategy
            waitStrategy, MessageConsumer[] messageConsumers) {

        //1. 构建ringBuffer对象 ，塞进去的对象类型
        this.ringBuffer = RingBuffer.create(type,
                TranslatorDataWapper::new,
                bufferSize,
                waitStrategy);

        //2.设置序号栅栏
        this.sequenceBarrier = this.ringBuffer.newBarrier();

        //3.设置工作池
        this.workerPool = new WorkerPool<TranslatorDataWapper>
                (this.ringBuffer, this.sequenceBarrier, new EventExceptionHandler(), messageConsumers);

        //4 把所构建的消费者置入池中
        for (MessageConsumer mc : messageConsumers) {
            consumers.put(mc.getConsumerId(), mc);
        }


        //5 添加我们的sequences
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());

        //6 启动我们的工作池
        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2));

    }

    public MessageProducer getMessageProducer(String producerId) {
        //池里有直接获取生产者
        MessageProducer messageProducer = this.producers.get(producerId);
        if(null == messageProducer) {
            messageProducer = new MessageProducer(producerId,
                    this.ringBuffer);
            this.producers.put(producerId, messageProducer);
        }
        return messageProducer;
    }

}
